package com.whcoding.task.xxl.job.task.base;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.Data;
import lombok.experimental.Accessors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.*;

/**
 * 多任务并发处理框架
 *
 * @description: 多线程并发处理框架
 * @author: ZhouWenHao
 * @date 2023/04/04 9:15
 */
@Data
public abstract class BaseTasker<T> {
    private static final Logger logger = LoggerFactory.getLogger(BaseTasker.class);

    @Data
    @Accessors(chain = true)
    public static final class Setting {
        /**
         * 线程配置
         */
        private int corePoolSize = 10;
        private int maximumPoolSize = 30;
        private long keepAliveTime = 5;
        private TimeUnit unit = TimeUnit.SECONDS;
        /**
         * 协调者休息时间 毫秒
         */
        private int coordinatorSleep = 1_000;
        /**
         * 协调者为任务分组的最小任务数
         */
        private int minTaskGroupSize = 50;
        /**
         * 协调者为任务分组的最大任务数
         */
        private int maxTaskGroupSize = 500;
        /**
         * 失败任务的延迟处理时间配置(秒)
         */
        private Integer[] retryDelaySeconds = new Integer[]{0, 3, 30, 5 * 60, 20 * 60, 60 * 60};
        /**
         * 是否打印指标数据
         */
        private boolean printMonitorInfo;
        /**
         * 打印指标数据间隔(秒)
         */
        private Integer printMonitorInterval = 60;
    }

    private Setting setting;
    private ThreadPoolExecutor executor;
    /**
     * 缓存队列 任务进来时全部进入此队列
     */
    private ConcurrentLinkedQueue<T> cacheQueue = new ConcurrentLinkedQueue<>();
    /**
     * 线程等待队列 线程池忙时 多余的任务在此等待
     */
    private BlockingQueue<Runnable> awaitQueue;
    /**
     * 重试等待队列 任务失败时 进入此队列等待重试
     */
    private BlockingQueue<Runnable> retryQueue;
    /**
     * 协调者
     */
    private Coordinator<T> coordinator;
    /**
     * 监控者
     */
    private Monitor monitor;

    public BaseTasker() {
        this(new Setting());
    }

    public BaseTasker(Setting setting) {
        this.awaitQueue = new LinkedBlockingQueue<>();

        this.executor = new ThreadPoolExecutor(
                setting.getCorePoolSize(),
                setting.getMaximumPoolSize(),
                setting.getKeepAliveTime(),
                setting.getUnit(),
                awaitQueue,
                new ThreadFactoryBuilder()
                        .setNameFormat("baseTasker-pool-%d").build());

        this.setting = setting;
        this.coordinator = new Coordinator<>(this);
        this.monitor = new Monitor(this);
    }

    /**
     * 对外提供的增加任务的接口(单)
     *
     * @param t 任务数据对象
     * @return boolean
     * @description 对外提供的增加任务的接口
     * @author whcoding
     * @date 2023/4/06 9:10
     */
    public boolean pushData(T t) {
        return cacheQueue.offer(t);
    }

    /**
     * 对外提供的增加任务的接口(多)
     *
     * @param list
     * @return boolean
     * @description 对外提供的增加任务的接口
     * @author whcoding
     * @date 2023/4/06 9:10
     */
    public boolean pushData(List<T> list) {
        if (CollectionUtils.isEmpty(list)) {
            return false;
        }
        return cacheQueue.addAll(list);
    }

    /**
     * 启动
     *
     * @param
     * @return void
     * @description 启动
     * @author whcoding
     * @date 2023/4/06 9:10
     */
    public void start() {
        coordinator.start();
    }

    /**
     * 任务实现
     *
     * @param list
     * @return void
     * @description 由继承者实现
     * @author whcoding
     * @date 2023/4/06 9:10
     */
    protected abstract void taskRunnable(List<T> list);

    /**
     * 异常数据的回调
     *
     * @param errorDataInfo
     */
    protected abstract void errorDataCallback(ErrorDataInfo<T> errorDataInfo);

}
